package com.anzhen.interview.concurrent;

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

/**
 * Created by anzhen on 2017-5-3.
 */
public class ForkJoinTest {
	public static void main(String[] args) {
		long[] longs = new long[400];
		for (int i = 0; i < longs.length; i++) {
			longs[i] = i;
		}

		ForkJoinPool forkJoinPool = new ForkJoinPool(4);
		SumTask sumTask = new SumTask(longs, 0, longs.length);
		long startTime = System.currentTimeMillis();
		Long result = forkJoinPool.invoke(sumTask);
		long endTime = System.currentTimeMillis();
		System.out.println("Fork/join sum: " + result + " in " + (endTime - startTime) + " ms.");

		ForkJoinPool forkJoinPool_ok = new ForkJoinPool(4);
		SumTask sumTask_ok = new SumTask(longs, 0, longs.length);
		long startTime_ok = System.currentTimeMillis();
		Long result_ok = forkJoinPool_ok.invoke(sumTask_ok);
		long endTime_ok = System.currentTimeMillis();
		System.out.println("Fork/join sum: " + result_ok + " in " + (endTime_ok - startTime_ok) + " ms.");

	}
}

class SumTask extends RecursiveTask<Long> {
	static final int THRESHOLD = 100;
	long[] array;
	int start;
	int end;

	SumTask(long[] array, int start, int end) {
		this.array = array;
		this.start = start;
		this.end = end;
	}

	/**
	 * 错误使用
	 * 执行compute()方法的线程本身也是一个Worker线程，当对两个子任务调用fork()时，这个Worker线程就会把任务分配给另外两个Worker，但是
	 * 它自己却停下来等待不干活了！这样就白白浪费了Fork/Join线程池中的一个Worker线程，导致了4个子任务至少需要7个线程才能并发执行。
	 * @return
	 */
	@Override
	protected Long compute() {
		if (end - start <= THRESHOLD) {
			// 如果任务足够小,直接计算:
			long sum = 0;
			for (int i = start; i < end; i++) {
				sum += array[i];
			}
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
			}
			System.out.println(Thread.currentThread().getName() + "---" + String.format("compute %d~%d = %d", start, end, sum));
			return sum;
		}
		int middle = (end + start) / 2;
		System.out.println(Thread.currentThread().getName() + "---" + String.format("split %d~%d ==> %d~%d, %d~%d", start, end, start, middle, middle, end));
		SumTask subtask1 = new SumTask(array,start,middle);
		SumTask subtask2 = new SumTask(array, middle, end);
		// 分别对子任务调用fork():
		subtask1.fork();
		subtask2.fork();
		// 合并结果:
		Long subresult1 = subtask1.join();
		Long subresult2 = subtask2.join();
		return subresult1 + subresult2;
	}
}

class SumTask_OK extends RecursiveTask<Long> {

	static final int THRESHOLD = 100;
	long[] array;
	int start;
	int end;

	SumTask_OK(long[] array, int start, int end) {
		this.array = array;
		this.start = start;
		this.end = end;
	}

	@Override
	protected Long compute() {
		if (end - start <= THRESHOLD) {
			// 如果任务足够小,直接计算:
			long sum = 0;
			for (int i = start; i < end; i++) {
				sum += array[i];
			}
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
			}
			System.out.println(String.format("compute %d~%d = %d", start, end, sum));
			return sum;
		}
		// 任务太大,一分为二:
		int middle = (end + start) / 2;
		System.out.println(String.format("split %d~%d ==> %d~%d, %d~%d", start, end, start, middle, middle, end));
		SumTask subtask1 = new SumTask(this.array, start, middle);
		SumTask subtask2 = new SumTask(this.array, middle, end);
		invokeAll(subtask1, subtask2);
		Long subresult1 = subtask1.join();
		Long subresult2 = subtask2.join();
		Long result = subresult1 + subresult2;
		System.out.println("result = " + subresult1 + " + " + subresult2 + " ==> " + result);
		return result;
	}
}
